package repeater

import (
    "fmt"
    "../common"
    "strconv"
)

type CarbonWriter struct {
    carbonAddr    string
    carbonConn    *common.Conn
    workingOffset int64
    rc            *RepeaterController
    batchSize     int
}

func NewCarbonWriter(cfg map[string]string) (*CarbonWriter, error) {
    addr, ok := cfg["carbon_addr"]
    if !ok {
        return nil, fmt.Errorf("config missing: [carbon_addr]")
    }
    batchSize := 10
    if b, ok := cfg["batch_size"]; ok {
        if s, err := strconv.Atoi(b); err == nil {
            batchSize = s
        }
    }
    return &CarbonWriter{
        carbonAddr:    addr,
        workingOffset: DEFAULT_OFFSET,
        batchSize:     batchSize,
    }, nil
}
func (cs *CarbonWriter) Init() error {
    conn, err := common.NewConn(cs.carbonAddr)
    if err != nil {
        return err
    }
    cs.carbonConn = conn
    return nil
}
func (cs *CarbonWriter) Start() {
    cs.rc.wg.Add(1)
}
func (cs *CarbonWriter) Stop() error {
    log.Debugf("carbon stop been called")
    defer cs.rc.waitGroup().Done()
    cs.carbonConn.Close()
    return nil
}
func (cs *CarbonWriter) Name() string {
    return "carbon" + cs.carbonAddr
}

func (cs *CarbonWriter) Handle(dps *DataPointPack) error {
    size := len(dps.Datapoints)
    for i:=0;i<size;i+=cs.batchSize{
        var n int
        var err error
        if next:=i+cs.batchSize;next>=size{
            n, err = cs.carbonConn.WriteDataPoints(dps.Datapoints[i:])
        }else{
            n, err = cs.carbonConn.WriteDataPoints(dps.Datapoints[i:next])
        }
        if err != nil {
            return err
        }
        if n == 0 {
            return fmt.Errorf("Zero Return Fom TCP")

        }
    }
    cs.workingOffset = dps.Offset()
    return nil
}

func (cs *CarbonWriter) SetController(rc *RepeaterController) error {
    cs.rc = rc
    return nil
}
func (cs *CarbonWriter) Commit() (int64, error) {
    if !cs.carbonConn.IsAlive() {
        return cs.workingOffset, fmt.Errorf("Carbon Conn Closed")
    }
    return cs.workingOffset, nil
}
